package cn.wangjie.spark.sources.rate

import org.apache.spark.sql.streaming.{OutputMode, StreamingQuery}
import org.apache.spark.sql.{DataFrame, SparkSession}

/**
 * 数据源：Rate Source，以每秒指定的行数生成数据，每个输出行包含一个timestamp和value。
 */
object StructuredRateSource {

  def main(args: Array[String]): Unit = {
		// 1. 创建SparkSession实例对象
		val spark: SparkSession = SparkSession
			.builder()
			.appName(this.getClass.getSimpleName.stripSuffix("$"))
			.master("local[2]")
			// 设置SparkSQL Shuffle时分区数目
			.config("spark.sql.shuffle.partitions", "2")
			.getOrCreate()
		// 导入隐式转换和函数库
		import spark.implicits._
		import org.apache.spark.sql.functions._
		
		// TODO: 2. Rate 数据源，自动生成数据
		val inputTable: DataFrame = spark.readStream
    		.format("rate")
			.option("rowsPerSecond", "10") // 每秒生成数据数目
			.option("rampUpTime", "0s") // 每条数据生成间隔时间
			.option("numPartitions", "2") // 分区数目
    		.load()
		inputTable.printSchema()
		
		// 3 输出数据和启动启用
		val query: StreamingQuery = inputTable
			.writeStream
			.outputMode(OutputMode.Append())
			.format("console")
			.option("numRows", "100")
			.option("truncate", "false")
			// 针对流式应用，需要启动执行
			.start()
		query.awaitTermination() // 当流式应用运行以后，正常情况一直运行，除非人为或程序异常终止，等待结束
		query.stop()
		
	}
	
}
